#include "socket_worker.h"

#include <sys/socket.h>
#include <fcntl.h>

#include <iostream>
#include <cassert>
#include <cstring>

#include "sunnet.h"
#include "socket_msg.h"


SocketWorker::SocketWorker(){
}

SocketWorker::~SocketWorker(){
}

void SocketWorker::OnEvent(epoll_event ev){
    LOG_DEBUG("OnEvent");
    int fd = ev.data.fd;
    auto conn = Sunnet::GetInstance().GetConn(fd);
    if(conn == nullptr){
        LOG_ERR("OnEvent error, conn == nullptr");
        return;
    }

    bool is_read = ev.events & EPOLLIN;
    bool is_write = ev.events & EPOLLOUT;
    bool is_error = ev.events & EPOLLERR;
    if(conn->GetType() == Conn::Type::LISTEN){
        if(is_read){
            OnAccept(conn);
        }
    }else{
        if(is_read || is_write){
            OnRW(conn, is_read, is_write);
        }
        if(is_error){
            LOG_WARN("OnError fd:{}", fd);
        }
    }
}

void SocketWorker::OnAccept(std::shared_ptr<Conn> conn){
    LOG_DEBUG("OnAccept");
    int cli_fd = accept(conn->GetFd(), nullptr, nullptr);
    if(cli_fd < 0){
        LOG_ERR("accept error");
        return;
    }

    fcntl(cli_fd, F_SETFL, O_NONBLOCK);
    Sunnet::GetInstance().AddConn(cli_fd, conn->GetServiceID(), Conn::Type::CLIENT);

    epoll_event ev;
    ev.events = EPOLLIN | EPOLLET;
    ev.data.fd = cli_fd;
    if(epoll_ctl(epfd_, EPOLL_CTL_ADD, cli_fd, &ev) == -1){
        LOG_ERR("accept epoll_ctl error:{}", strerror(errno));
    }

    auto msg = std::make_shared<SocketAcceptMsg>();
    msg->SetType(BaseMsg::MsgType::SOCKET_ACCEPT);
    msg->SetListenFd(conn->GetFd());
    msg->SetCliFd(cli_fd);
    Sunnet::GetInstance().Send(conn->GetServiceID(), msg);
}

void SocketWorker::OnRW(std::shared_ptr<Conn> conn, bool r, bool w){
    LOG_DEBUG("OnRW, fd:{} r:{} w:{}", conn->GetFd(), r, w);
    auto msg = std::make_shared<SocketRWMsg>();
    msg->SetType(BaseMsg::MsgType::SOCKET_RW);
    msg->SetRead(r);
    msg->SetWrite(w);
    msg->SetFd(conn->GetFd());
    Sunnet::GetInstance().Send(conn->GetServiceID(), msg);
}

void SocketWorker::Init(){
    LOG_INFO("socket worker init");
    epfd_ = epoll_create(1);
    assert(epfd_ > 0);
}

void SocketWorker::operator()(){
    LOG_INFO("socket worker start");
    while(true){
        const int EVENT_SIZE = 64;
        epoll_event events[EVENT_SIZE];
        int ev_count = epoll_wait(epfd_, events, EVENT_SIZE, -1);
        for(int i = 0; i < ev_count; ++i){
            epoll_event ev = events[i];
            OnEvent(ev);
        }
    }
}

void SocketWorker::AddEvent(int fd){
    LOG_DEBUG("AddEvent fd:{}", fd);
    epoll_event ev;
    ev.events = EPOLLIN | EPOLLET;
    ev.data.fd = fd;
    if(epoll_ctl(epfd_, EPOLL_CTL_ADD, fd, &ev) == -1){
        LOG_ERR("AddEvent error: {}", strerror(errno));
    }
}
void SocketWorker::RemoveEvent(int fd){
    LOG_DEBUG("RemoveEvent fd:{}", fd);
    epoll_ctl(epfd_, EPOLL_CTL_DEL, fd, nullptr);
}
void SocketWorker::ModifyEvent(int fd, bool epoll_out){
    LOG_DEBUG("ModifyEvent fd:{} epoll_out:{}", fd, epoll_out);
    epoll_event ev;
    ev.data.fd = fd;
    if(epoll_out){
        ev.events = EPOLLIN | EPOLLET | EPOLLOUT;
    }else{
        ev.events = EPOLLIN | EPOLLET;
    }
    epoll_ctl(epfd_, EPOLL_CTL_MOD, fd, &ev);
}